package com.foreveross.taskservice.engine.worker;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

import org.dayatang.domain.InstanceFactory;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.UnableToInterruptJobException;

import com.foreveross.taskservice.application.channel.IGradChannelApplication;
import com.foreveross.taskservice.application.impl.convert.AirFreightChannelConvert;
import com.foreveross.taskservice.application.task.statsdetail.vo.BaseChannelVO;
import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.TaskStatusEnum;
import com.foreveross.taskservice.common.model.AbstractAction;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.common.model.airfreight.AirFreightTaskModel;
import com.foreveross.taskservice.core.domain.channel.GradChannelEntity;
import com.foreveross.taskservice.infra.entity.DetailInfoEntity;
import com.foreveross.taskservice.infra.entity.StaticHistoryHbaseEntity;
import com.foreveross.taskservice.taskcache.ITaskCacheQuery;
import com.google.common.collect.Lists;
public class HistoryDetailHbaseWorker extends AbstractTaskStatefulJobWorker{
	
	
	public void saveHbase()throws Exception{
		//查询出数据（详情） 
		List<AbstractTaskModel> tasks=new ArrayList<AbstractTaskModel>();
		String grabDateStr=new SimpleDateFormat("yyyy-MM-dd").format(new Date());
		for(GradChannelEntity channel:getAllChannels()){
			log.info(String.format("【%s持久化...】", channel.getName()));
//			long channelId=channel.getId();
			String channelId=channel.getHbaseKey();
			int flage=0;
			for(int a=0;a<TaskStatusEnum.values().length;a++){
				log.info(String.format("【%s持久化...】",TaskStatusEnum.values()[a].getRemake()));
				int status=TaskStatusEnum.values()[a].getStatus();
//				if(status==0)continue;
				//根据条件查询redis
				//处理内存溢出：分段入库
				int  beginInt=0,f=10000,flagInt=0;
				while(tasks.size()>=beginInt){ 
					tasks=new ArrayList<AbstractTaskModel>();
					tasks=InstanceFactory.getInstance(ITaskCacheQuery.class).queryTaskDetailInfo(CacheDataSpace.AirFreightSpace(), channelId,TaskStatusEnum.getTaskStatusEnum(status),flagInt,f);
					//填写entity，使用Hibernate映射
					this.saveEntity(tasks, channel, status, grabDateStr);
					beginInt=f+1;
					flagInt=flagInt+tasks.size()+1;
				}
				flage++;
			}
			
			/*//单独处理“没有航班信息”数据
			if(flage==TaskStatusEnum.values().length){
				tasks=InstanceFactory.getInstance(ITaskCacheQuery.class).queryTaskDetailInfo(CacheDataSpace.AirFreightSpace(), channelId,TaskStatusEnum.getTaskStatusEnum(6),0,999999);
				this.saveEntity(tasks, channel, 6, grabDateStr, listEntity);
			}*/
		}
	}
	
	
	
	//保存 Entity
	public void saveEntity(List<AbstractTaskModel> tasks,GradChannelEntity channel,int status,String grabDateStr){
		if(tasks.size()>0){
			BaseChannelVO channelVO=new BaseChannelVO();
			List<StaticHistoryHbaseEntity> listEntity=new ArrayList<StaticHistoryHbaseEntity>();
			for(int i=0;i<tasks.size();i++){
				AirFreightTaskModel vo=(AirFreightTaskModel) tasks.get(i);
				channelVO.setChannelModel(AirFreightChannelConvert.convert(channel));
				channelVO.setChannelId(channel.getId());
				channelVO.setName(channel.getName());
				StaticHistoryHbaseEntity en=new StaticHistoryHbaseEntity();
				Set<DetailInfoEntity> de=new LinkedHashSet<DetailInfoEntity>();
				if(vo.getActions().size()>0){
					for(AbstractAction ac:vo.getActions()){
						DetailInfoEntity ded=new DetailInfoEntity();
						ded.setActionType(ac.getActionType());
						ded.setRemark(ac.getRemark());
						ded.setTimestamp(ac.getTimestamp());
						ded.setTimestampStr(ac.getTimestampStr());
						de.add(ded);
					}
				}
				en.setGrabChannelCode(channel.getHbaseKey());
				en.setFromCityName(vo.getFromCityName());
				en.setFromCity(vo.getFromCity());
				en.setToCityName(vo.getToCityName());
				en.setToCity(vo.getToCity());
				en.setTaskId(vo.getId());
				en.setExcuteCount(vo.getExcuteCount());
				en.setPriority(vo.getPriority());
				en.setCrawlCount(vo.getCrawlCount());
				en.setGrabChannelId(channelVO.getChannelId());
				en.setStartTime(vo.getStartTime());
				en.setEndTime(vo.getEndTime());
				en.setFlightDate(vo.getFlightDate());
				en.setReturnFlightDate(vo.getReturnGrabDate());
				en.setGrabChannelName(channelVO.getChannelModel().getName());
				en.setStatus(status);
				en.setGrabDate(grabDateStr);
				if(de.size()>0){
					en.setDetails(de);
				}
				listEntity.add(en);
			}
			
			//循环插入
			int i=0;
			for(StaticHistoryHbaseEntity entity:listEntity){
				log.info(String.format("进库【%s】【%s/%s】",channel.getName(),i,listEntity.size()));
				staticHistoryApplication.save(entity);
				i++;
			}
		}
		
		
	}

	
	//获取所有的渠道
	public List<GradChannelEntity> getAllChannels() throws Exception{
		List<GradChannelEntity> channels=Lists.newArrayList();
		channels=InstanceFactory.getInstance(IGradChannelApplication.class).getEnableChannel();
		return channels;
	}
	
	public void interrupt() throws UnableToInterruptJobException {
		// TODO Auto-generated method stub
		
	}

	public void execute(JobExecutionContext context)
			throws JobExecutionException {
		// TODO Auto-generated method stub
		
	}
}
